package com.atguigu.edu.realtime220815.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime220815.app.func.DimAsyncFunction;
import com.atguigu.edu.realtime220815.bean.ExamPaperBean;
import com.atguigu.edu.realtime220815.bean.ExamPaperIntervalScoreBean;
import com.atguigu.edu.realtime220815.util.ClickHouseUtils;
import com.atguigu.edu.realtime220815.util.DateFormatUtil;
import com.atguigu.edu.realtime220815.util.KafkaUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.Collections;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;

/**
 * @Author: rym
 * @Description: TODO
 * @DateTime: 2023/2/18 16:27
 **/
public class DwsExamPaperIntervalScoreWindow {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);

        String topic = "dwd_exam_user_detail";
        String groupId = "DwsExamPaperIntervalScoreWindow_group";
        FlinkKafkaConsumer<String> flinkKafkaConsumer = KafkaUtils.getFlinkKafkaConsumer(topic, groupId);
        DataStreamSource<String> kfkDS = env.addSource(flinkKafkaConsumer);

        SingleOutputStreamOperator<JSONObject> jsonObjDS = kfkDS.map(JSON::parseObject);

        //{"id":"31595","paper_id":"168","user_id":"968","score":"16.7","duration_sec":"1084","date_id":"2023-02-18","ts":"1676700668"}
        SingleOutputStreamOperator<ExamPaperIntervalScoreBean> beanDS = jsonObjDS.map(
                new MapFunction<JSONObject, ExamPaperIntervalScoreBean>() {
                    @Override
                    public ExamPaperIntervalScoreBean map(JSONObject value) throws Exception {
                        String user_id = value.getString("user_id");
                        String paper_id = value.getString("paper_id");
                        Double score = value.getDouble("score");
                        Long ts = value.getLong("ts");
                        if (score > 80.0) {
                            ExamPaperIntervalScoreBean A = ExamPaperIntervalScoreBean.builder()
                                    .setUid(new HashSet<>(Collections.singleton(user_id)))
                                    .ts(ts)
                                    .scoreType("A")
                                    .paper_id(paper_id)
                                    .build();
                            return A;
                        }
                        if (score > 60.0 && score <= 80.0) {
                            ExamPaperIntervalScoreBean B = ExamPaperIntervalScoreBean.builder()
                                    .setUid(new HashSet<>(Collections.singleton(user_id)))
                                    .ts(ts)
                                    .scoreType("B")
                                    .paper_id(paper_id)
                                    .build();
                            return B;
                        } else {
                            ExamPaperIntervalScoreBean C = ExamPaperIntervalScoreBean.builder()
                                    .setUid(new HashSet<>(Collections.singleton(user_id)))
                                    .ts(ts)
                                    .scoreType("C")
                                    .paper_id(paper_id)
                                    .build();
                            return C;
                        }


                    }
                }
        );

        SingleOutputStreamOperator<ExamPaperIntervalScoreBean> withWaterDS = beanDS.assignTimestampsAndWatermarks(WatermarkStrategy.<ExamPaperIntervalScoreBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                .withTimestampAssigner(new SerializableTimestampAssigner<ExamPaperIntervalScoreBean>() {
                    @Override
                    public long extractTimestamp(ExamPaperIntervalScoreBean element, long recordTimestamp) {
                        return element.getTs()*1000;
                    }
                }));

        KeyedStream<ExamPaperIntervalScoreBean, String> keyDS = withWaterDS.keyBy(value -> value.getPaper_id());

        WindowedStream<ExamPaperIntervalScoreBean, String, TimeWindow> windowDS = keyDS.window(TumblingProcessingTimeWindows.of(Time.seconds(10)));

        SingleOutputStreamOperator<ExamPaperIntervalScoreBean> reduceDS = windowDS.reduce(
                new ReduceFunction<ExamPaperIntervalScoreBean>() {
                    @Override
                    public ExamPaperIntervalScoreBean reduce(ExamPaperIntervalScoreBean value1, ExamPaperIntervalScoreBean value2) throws Exception {
                        value1.getSetUid().addAll(value2.getSetUid());
                        return value1;
                    }
                },
                new WindowFunction<ExamPaperIntervalScoreBean, ExamPaperIntervalScoreBean, String, TimeWindow>() {
                    @Override
                    public void apply(String s, TimeWindow window, Iterable<ExamPaperIntervalScoreBean> input, Collector<ExamPaperIntervalScoreBean> out) throws Exception {
                        long start = window.getStart();
                        long end = window.getEnd();
                        long now = System.currentTimeMillis();

                        for (ExamPaperIntervalScoreBean examPaperIntervalScoreBean : input) {
                            examPaperIntervalScoreBean.setStt(DateFormatUtil.toYmdHms(start));
                            examPaperIntervalScoreBean.setEdt(DateFormatUtil.toYmdHms(end));
                            examPaperIntervalScoreBean.setTs(now);
                            examPaperIntervalScoreBean.setPeople_counting((long) examPaperIntervalScoreBean.getSetUid().size());
                            out.collect(examPaperIntervalScoreBean);
                        }
                    }
                }
        );

        SingleOutputStreamOperator<ExamPaperIntervalScoreBean> withPaperIdDS = AsyncDataStream.unorderedWait(
                reduceDS,
                new DimAsyncFunction<ExamPaperIntervalScoreBean>("dim_test_paper") {
                    @Override
                    public void join(JSONObject dimInfoJsonObj, ExamPaperIntervalScoreBean obj) {
                        String paper_title = dimInfoJsonObj.getString("paper_title".toUpperCase());
                        obj.setPaper_name(paper_title);
                    }

                    @Override
                    public String getKey(ExamPaperIntervalScoreBean obj) {
                        return obj.getPaper_id();
                    }
                }, 60, TimeUnit.SECONDS
        );

        withPaperIdDS.addSink(ClickHouseUtils.getClickHouseSinkFunction(
                "insert into dws_exam_paper_interval_score_window values(?,?,?,?,?,?,?)"
        ));
        env.execute();
    }
}
